package com.github.binarylei.security.topology;

import com.github.binarylei.security.bolt.WorldSplitBolt;
import com.github.binarylei.security.bolt.WorldWriteBolt;
import com.github.binarylei.security.spout.WorldSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;

/**
 * Storm 安全机制，数据丢失可心在spout中重新发送，防止数据丢失
 * 观察在 WorldSplitBolt 中出现异常和在 WorldWriteBolt 中出现异常的结果
 * @author: leigang
 * @version: 2018-04-07
 */
public class SecurityTopology {

    public static void main(String[] args) throws InterruptedException {

        WorldSpout spout = new WorldSpout();
        WorldSplitBolt splitBolt = new WorldSplitBolt();
        WorldWriteBolt writeBolt = new WorldWriteBolt();

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("spout", spout);
        builder.setBolt("splitBolt", splitBolt, 5)
                .shuffleGrouping("spout");
        builder.setBolt("writeBolt", writeBolt, 5)
                .globalGrouping("splitBolt");

        Config cfg = new Config();
        //cfg.setNumWorkers(1);
        cfg.setDebug(true);

        //1. 本地模式
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("topo", cfg, builder.createTopology());
        Thread.sleep(20 * 1000);
        cluster.killTopology("topo");
        cluster.shutdown();

        //2. 集群模式
        //cfg.setNumWorkers(1);
        //cfg.setDebug(true);
        //StormSubmitter.submitTopology("top1", cfg, builder.createTopology());

    }
}
